package org.lk.controller;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import org.lk.common.FlinkCdcStreamManager;
import org.lk.common.log.CdcFileAppender;
import org.lk.common.log.LogResult;
import org.lk.entity.CdcJob;
import org.lk.entity.CdcJobTable;
import org.lk.entity.CdcLog;
import org.lk.mapper.CdcJobMapper;
import org.lk.mapper.CdcJobTableMapper;
import org.lk.mapper.CdcLogMapper;
import org.lk.param.LogParam;
import org.lk.util.ResponseResult;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;

/**
 * @author: lk
 * @since:
 */
@RestController
@RequestMapping("cdc")
public class FlinkCdcController {
    @Resource
    private FlinkCdcStreamManager flinkCdcStreamManager;

    @Resource
    private CdcJobMapper cdcJobMapper;
    @Resource
    private CdcJobTableMapper cdcJobTableMapper;
    @Resource
    private CdcLogMapper cdcLogMapper;

    @GetMapping("start")
    public ResponseResult start(String id,String jobId) {
        CdcJob cdcJob = cdcJobMapper.selectById(id);
        QueryWrapper<CdcJobTable> query = Wrappers.<CdcJobTable>query()
                .eq("cdc_id", cdcJob.getId());
        List<CdcJobTable> cdcJobTables = cdcJobTableMapper.selectList(query);
        flinkCdcStreamManager.setCdcJob(cdcJob);
        flinkCdcStreamManager.setCdcJobTableList(cdcJobTables);
        flinkCdcStreamManager.setJobId(jobId);
        flinkCdcStreamManager.start();
        return ResponseResult.success();
    }


    @GetMapping("stop")
    public ResponseResult stop(String id) {
        CdcJob cdcJob = cdcJobMapper.selectById(id);
        return flinkCdcStreamManager.stop(cdcJob.getJobName());
    }


    @GetMapping("logRead")
    public ResponseResult<LogResult> logRead(long logId, int fromLineNum) {
        try {
            //根据日志ID 从数据库中查询日志ID
            CdcLog cdcLog = cdcLogMapper.selectById(logId);
            LogParam logParam = new LogParam(cdcLog.getTriggerTime().getTime(), cdcLog.getId(), fromLineNum);
            String logFileName = CdcFileAppender.makeLogFileName(new Date(logParam.getLogDateTime()), logParam.getLogId());
            // 开始读取到日志文件
            LogResult logResult = CdcFileAppender.readLog(logFileName, logParam.getFromLineNum());
            // 读取到最后
            if (logResult.getLogContent() != null && logResult.getFromLineNum() > logResult.getToLineNum()) {
                if (cdcLog.getHandlerCode() > 0) {
                    logResult.setEnd(true);
                }
            }
            return ResponseResult.success(logResult);
        } catch (Exception e) {
            e.printStackTrace();
            return ResponseResult.fail(e.getMessage());
        }

    }


}
